package org.joa.actor.transaction;

import java.io.IOException;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.logging.Logger;

import org.joa.actor.Actor;
import org.joa.actor.ActorName;
import org.joa.actor.ReadOnly;
import org.joa.actor.exception.RemoteException;
import org.joa.manager.ActorManager;

public class TransactionActor extends Actor implements Serializable {
    public static final long READ_REVISION_CURRENT = -1;

    public static final int MAX_RETRY_COUNT = 10;

    private static final long serialVersionUID = 1L;

    private static class ActorTransactionInfo implements Serializable {
        private static final long serialVersionUID = 1L;

        private long pendingMessageCount = 0;

        @SuppressWarnings("unused")
        public long getPendingMessageCount() {
            return pendingMessageCount;
        }

        public void incrementPendingMessageCount() {
            pendingMessageCount++;
        }

        public void decrementPendingMessageCount() {
            pendingMessageCount--;
        }
    }

    private Map<ActorName, ActorTransactionInfo> actorTransactionInfoMap =
        new HashMap<ActorName, ActorTransactionInfo>();

    private long readRevision = READ_REVISION_CURRENT;

    private long pendingMessageCount = 0;
    private transient ActorManager actorManager = ActorManager.instance;
    private transient Logger logger = Logger.getLogger(TransactionActor.class.getName());

    private boolean retry = false;
    private Callable<?> retryCallable;
    private int retryCount = 0;

    private boolean rollback = false;

    public TransactionActor() {
    }

    public TransactionActor(Callable<?> retryCallable) {
        this.retryCallable = retryCallable;
    }

    @ReadOnly
    public long getReadRevision() {
        return readRevision;
    }

    protected void commit() {
        long revision = actorManager.getCurrentRevision() + 1;
        for (ActorName actorName : actorTransactionInfoMap.keySet()) {
            try {
                actorManager.commit(actorName, revision);
            } catch (IOException e) {
                // TODO: Handle this case
                e.printStackTrace();
            }
        }
    }

    public void retry() throws IOException {
        retry = true;
    }

    public void rollback() throws IOException {
        rollback = true;
    }

    public void incrementPendingMessageCount(ActorName actorName) {
        ActorTransactionInfo transactionInfo = getTransactionInfo(actorName);
        transactionInfo.incrementPendingMessageCount();

        pendingMessageCount++;

        logger.fine("Transaction: " + self() + " incrementPendingMessageCount(" + actorName +
            ") - pendingMessageCount: " + pendingMessageCount);
    }

    public void decrementPendingMessageCount(ActorName actorName) throws RemoteException {
        ActorTransactionInfo transactionInfo = getTransactionInfo(actorName);
        transactionInfo.decrementPendingMessageCount();

        pendingMessageCount--;

        logger.fine("Transaction: " + self() + " decrementPendingMessageCount(" + actorName +
            ") - pendingMessageCount: " + pendingMessageCount);

        if (pendingMessageCount == 0) {
            onPendingMessagesComplete();
        }
    }

    public void setActorTransactionInfo(ActorName actorName, ActorTransactionInfo transactionInfo) {
        actorTransactionInfoMap.put(actorName, transactionInfo);
    }

    protected ActorTransactionInfo getTransactionInfo(ActorName actorName) {
        ActorTransactionInfo transactionInfo = actorTransactionInfoMap.get(actorName);
        if (transactionInfo == null) {
            transactionInfo = new ActorTransactionInfo();
            actorTransactionInfoMap.put(actorName, transactionInfo);
        }

        return transactionInfo;
    }

    protected void onPendingMessagesComplete() throws RemoteException {
        if (retry) {
            executeRetry();
        } else if (rollback) {
            executeRollback();
        } else {
            commit();
        }
    }

    protected void executeRetry() throws RemoteException {
        executeRollback();

        if (retryCallable != null && retryCount < MAX_RETRY_COUNT) {
            try {
                retryCount++;

                retrySleep();

                retryCallable.call();
            } catch (Exception e) {
                throw new RemoteException(e);
            }
        }
    }

    protected void executeRollback() throws RemoteException {
        for (ActorName actorName : actorTransactionInfoMap.keySet()) {
            actorManager.rollback(actorName);
        }

        pendingMessageCount = 0;
        actorTransactionInfoMap.clear();
        retry = false;
        rollback = false;
    }

    protected void retrySleep() throws InterruptedException {
        Random randomGenerator = new Random();
        Thread.sleep(randomGenerator.nextInt(retryCount * 250));
    }
}
